# Databricks notebook source
# MAGIC %md
# MAGIC # Read and process original data (build new pk)

# COMMAND ----------

from pyspark.sql.functions import col, lit, concat

df_interopko=spark.table("admin_govern_sta_des.tmp_mostlyai.interopko")
df_interopko=(df_interopko
         .withColumn("pk", concat(col("Usuari_solicitant"), lit("_"),col("N_Expedient"), lit("_"), col("N_Procediment"))))
display(df_interopko.limit(5))

# COMMAND ----------

df_interopok = spark.table("admin_govern_sta_des.tmp_mostlyai.interopok")
df_interopok=(df_interopok
         .withColumn("pk", concat(col("Usuari_solicitant"), lit("_"),col("N_Expedient"), lit("_"), col("N_Procediment")))
         )
display(df_interopok)

# COMMAND ----------

df_fullquery = spark.table("admin_govern_sta_des.tmp_mostlyai.fullquery")
df_fullquery=(df_fullquery
         .withColumn("pk", concat(col("identificador"), lit("_"),col("N_Expedient"), lit("_"), col("N_Procediment"))))
display(df_fullquery.limit(20))

# COMMAND ----------

from pyspark.sql.functions import col, regexp_replace, cast, lit, concat
df_pagamentnomina=spark.table("admin_govern_sta_des.tmp_mostlyai.pagamentnomina")
df_pagamentnomina=(df_pagamentnomina
                   .withColumn("Import", regexp_replace(regexp_replace(col("Import"), "\.", ""), ",", ".").cast("double"))
                   .withColumn("pk", concat(col("Usuari_solicitant"), lit("_"),col("N_Expedient"), lit("_"), col("N_Procediment"))))
display(df_pagamentnomina.limit(5))

# COMMAND ----------

from pyspark.sql.functions import concat, col, lit

df_actuacionsnomina_raw=spark.table("admin_govern_sta_des.tmp_mostlyai.actuacionsnomina")
df_fullquery_=spark.table("admin_govern_sta_des.tmp_mostlyai.fullquery").select("identificador", "N_Expedient", "N_Procediment").distinct()

df_actuacionsnomina = (
    df_actuacionsnomina_raw.join(df_fullquery_, on=["N_Expedient", "N_Procediment"], how="left")
    .withColumn("pk", concat(col("identificador"), lit("_"),col("N_Expedient"), lit("_"), col("N_Procediment")))
    .select("pk", "identificador", "N_Expedient", "N_Procediment", "Data_actuacio", "Efecte_Nomina")
)

display(df_actuacionsnomina.limit(5))

# COMMAND ----------

from pyspark.sql.functions import col, lit, concat

df_interopko_renamed = df_interopko.withColumnRenamed('Usuari_solicitant', 'identificador')
df_interopok_renamed = df_interopok.withColumnRenamed('Usuari_solicitant', 'identificador')
df_pagamentnomina_renamed = df_pagamentnomina.withColumnRenamed('Usuari_solicitant', 'identificador')

df_relationship = (
  df_interopko_renamed.select(['identificador', 'N_Expedient', 'N_Procediment'])
            .union(df_interopok_renamed.select(['identificador', 'N_Expedient', 'N_Procediment']))
            .union(df_fullquery.select(['identificador', 'N_Expedient', 'N_Procediment']))
            .union(df_pagamentnomina_renamed.select(['identificador', 'N_Expedient', 'N_Procediment'])) 
            .select("*")
            .distinct()
            .withColumn("pk", concat(col("identificador"), lit("_"),col("N_Expedient"), lit("_"), col("N_Procediment")))
            .withColumn("fk", concat(col("identificador"), lit("_"),col("N_Expedient")))
            .select("pk", "fk", "N_Procediment", "identificador")
            )
display(df_relationship.select("pk").distinct())

# COMMAND ----------

from pyspark.sql.functions import col, lit, concat
df_solicitud=spark.table("admin_govern_sta_des.tmp_mostlyai.solicitud")
df_solicitud=(df_solicitud
              .drop("_c166")
              .withColumn("fk", concat(col("solicitant1Identificador"), lit("_"),col("expedient"))))
display(df_solicitud.limit(5))

# COMMAND ----------

df_titular=(
  spark.table("admin_govern_sta_des.tmp_mostlyai.titular")
    .withColumnRenamed("identificador", "solicitant1Identificador")
            )
display(df_titular.limit(5))

# COMMAND ----------

df_esborrany = spark.table("admin_govern_sta_des.tmp_mostlyai.esborrany").withColumnRenamed("identificador_document", "solicitant1Identificador")
display(df_esborrany)

# COMMAND ----------

# MAGIC %md
# MAGIC # Read and process synthetic data (fetch cols from relationship)

# COMMAND ----------

df_synthetic_actuacionsnomina = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_actuacionsnomina_01_04")
df_synthetic_esborrany = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_esborrany_01_04")
df_synthetic_fullquery = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_fullquery_01_04")
df_synthetic_interopko = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_interopko_01_04")
df_synthetic_interopok = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_interopok_01_04")
df_synthetic_solicitud = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_solicitud_01_04")
df_synthetic_pagamentnomina = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_pagamentnomina_01_04")
df_synthetic_titular = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_titular_01_04")

df_synthetic_relationship = spark.table("admin_govern_sta_des.tmp_mostlyai.synthetic_relationship_01_04")

# COMMAND ----------

df_synthetic_relationship_joined = df_synthetic_relationship.join(df_synthetic_solicitud.select("fk", "expedient"), on=['fk'], how='left')
display(df_synthetic_relationship_joined)

# COMMAND ----------

df_synthetic_interopok_joined = df_synthetic_interopok.join(df_synthetic_relationship_joined.select("pk", "identificador", "N_Procediment", "expedient"), on=['pk'], how='left')
display(df_synthetic_interopok_joined)

# COMMAND ----------

df_synthetic_interopok_joined = df_synthetic_interopok.join(df_synthetic_relationship_joined.select("pk", "identificador", "N_Procediment", "expedient"), on=['pk'], how='left')
display(df_synthetic_interopok_joined)

# COMMAND ----------

df_synthetic_interopko_joined = df_synthetic_interopko.join(df_synthetic_relationship_joined.select("pk", "identificador", "N_Procediment", "expedient"), on=['pk'], how='left')
display(df_synthetic_interopko_joined)

# COMMAND ----------

df_synthetic_fullquery_joined = df_synthetic_fullquery.join(df_synthetic_relationship_joined.select("pk", "identificador", "N_Procediment", "expedient"), on=['pk'], how='left')
display(df_synthetic_fullquery_joined)

# COMMAND ----------

df_synthetic_pagamentnomina_joined = df_synthetic_pagamentnomina.join(df_synthetic_relationship_joined.select("pk", "identificador", "N_Procediment", "expedient"), on=['pk'], how='left')
display(df_synthetic_pagamentnomina_joined)

# COMMAND ----------

df_synthetic_actuacionsnomina_joined = df_synthetic_actuacionsnomina.join(df_synthetic_relationship_joined.select("pk", "identificador", "N_Procediment", "expedient"), on=['pk'], how='left')
display(df_synthetic_actuacionsnomina_joined)

# COMMAND ----------

# MAGIC %md
# MAGIC # Example

# COMMAND ----------

display(df_pagamentnomina_renamed.limit(10))

# COMMAND ----------

display(df_synthetic_pagamentnomina_joined.limit(10))

# COMMAND ----------

# MAGIC %md
# MAGIC # Count rows

# COMMAND ----------

tables = [
    df_actuacionsnomina,
    df_esborrany,
    df_fullquery,
    df_interopko,
    df_interopok,
    df_solicitud,
    df_pagamentnomina,
    df_titular,
    df_relationship
]

for df in tables:
    print(f"Table has {df.count()} rows")

# COMMAND ----------

tables = [
    df_synthetic_actuacionsnomina,
    df_synthetic_esborrany,
    df_synthetic_fullquery,
    df_synthetic_interopko,
    df_synthetic_interopok,
    df_synthetic_solicitud,
    df_synthetic_pagamentnomina,
    df_synthetic_titular,
    df_synthetic_relationship
]

for df in tables:
    print(f"Table has {df.count()} rows")

# COMMAND ----------

# MAGIC %md
# MAGIC # Checking cardinalities

# COMMAND ----------

# MAGIC %md
# MAGIC ## Actuacions Nomina

# COMMAND ----------

# MAGIC %md
# MAGIC ### Original

# COMMAND ----------

df_grouped = df_actuacionsnomina.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count()
df = df_grouped.groupBy("count").count()
display(df)

# COMMAND ----------

df_grouped = df_synthetic_actuacionsnomina.groupBy(["pk"]).count()
df = df_grouped.groupBy("count").count()
display(df)

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_actuacionsnomina.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_actuacionsnomina_joined.groupBy(["identificador", "N_Procediment", "expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Pagament nomina

# COMMAND ----------

# MAGIC %md
# MAGIC ### Original

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_pagamentnomina.groupBy(["Usuari_solicitant", "N_Procediment", "N_Expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_pagamentnomina_joined.groupBy(["identificador", "N_Procediment", "expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Full Query

# COMMAND ----------

# MAGIC %md
# MAGIC ### Original

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_fullquery.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_fullquery_joined.groupBy(["identificador", "N_Procediment", "expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Interopko

# COMMAND ----------

# MAGIC %md
# MAGIC ### Original

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_interopko.groupBy(["Usuari_solicitant", "N_Procediment", "N_Expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_interopko_joined.groupBy(["identificador", "N_Procediment", "expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Interopok

# COMMAND ----------

# MAGIC %md
# MAGIC ### Original

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_interopok.groupBy(["Usuari_solicitant", "N_Procediment", "N_Expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_interopok_joined.groupBy(["identificador", "N_Procediment", "expedient"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Solicituds

# COMMAND ----------

# MAGIC %md
# MAGIC ### Original

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_solicitud.groupBy(["expedient", "solicitant1Identificador"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_solicitud.groupBy(["expedient", "solicitant1Identificador"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ## Titular

# COMMAND ----------

# MAGIC %md
# MAGIC ###Original

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_titular.groupBy(["solicitant1identificador"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC ### Synthetic

# COMMAND ----------

from pyspark.sql.functions import avg, median

df_grouped = df_synthetic_titular.groupBy(["solicitant1identificador"]).count()
df_avg_repetitions = df_grouped.agg(avg("count").alias("avg_repetitions"), median("count").alias("median_repetitions"))
display(df_avg_repetitions)

# COMMAND ----------

# MAGIC %md
# MAGIC # Further analysis

# COMMAND ----------

df_grouped_actuacionsnomina = df_actuacionsnomina.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_actuacionsnomina)

# COMMAND ----------

df_grouped_synthetic_actuacionsnomina = df_synthetic_actuacionsnomina.groupBy(["pk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_actuacionsnomina)

# COMMAND ----------

df_grouped_pagamentnomina = df_pagamentnomina_renamed.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_pagamentnomina)

# COMMAND ----------

df_grouped_synthetic_pagamentnomina = df_synthetic_pagamentnomina.groupBy(["pk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_pagamentnomina)

# COMMAND ----------

df_grouped_fullquery = df_fullquery.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_fullquery)

# COMMAND ----------

df_grouped_synthetic_fullquery = df_synthetic_fullquery.groupBy(["pk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_fullquery)

# COMMAND ----------

df_grouped_interopok = df_interopok_renamed.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_interopok)

# COMMAND ----------

df_grouped_synthetic_interopok = df_synthetic_interopok.groupBy(["pk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_interopok)

# COMMAND ----------

df_grouped_interopko = df_interopko_renamed.groupBy(["identificador", "N_Procediment", "N_Expedient"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_interopko)

# COMMAND ----------

df_grouped_synthetic_interopko = df_synthetic_interopko.groupBy(["pk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_interopko)

# COMMAND ----------

display(df_solicitud)

# COMMAND ----------

df_grouped_solicitud = df_solicitud.groupBy(["fk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_solicitud)

# COMMAND ----------

df_grouped_synthetic_solicitud = df_synthetic_solicitud.groupBy(["fk"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_solicitud)

# COMMAND ----------

df_grouped_titular = df_titular.groupBy(["solicitant1identificador"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_titular)

# COMMAND ----------

df_grouped_synthetic_titular = df_synthetic_titular.groupBy(["solicitant1identificador"]).count().withColumnRenamed("count", "n_repetitions").groupBy("n_repetitions").count().withColumnRenamed("count", "n_pks").orderBy("n_repetitions")
display(df_grouped_synthetic_titular)

# COMMAND ----------



# COMMAND ----------

# df = df_grouped.groupBy("count").count()
# display(df)